//
//  rtmp_sender_impl.h
//  RtmpSender
//
//  Created by LSQ on 2020/2/12.
//  Copyright © 2020 Agora. All rights reserved.
//
#pragma once

#include <chrono>
#include <list>
#include <mutex>
#include <string>
#include <thread>
#ifndef WIN32
#include <sys/time.h>
#include <unistd.h>
#endif
#ifdef ANDROID
#include <asm/fcntl.h>
#include <semaphore.h>
#elif defined(WIN32)
#include <Windows.h>
#else
#include <fcntl.h>
#endif

#include "facilities/tools/rtc_callback.h"
#include "rtmp_sender.h"
#include "rtmp_streaming_buffer.h"
#include "third_party/srs-librtmp/src/srs/srs_librtmp.h"

namespace agora {
namespace rtc {

class RtmpSenderImpl : public IAgoraRtmpSender, IRtmpStreamingBufferObserver {
 public:
  using task_type = commons::async_queue_base::task_type;
  using async_queue_type = commons::async_queue_base;

 public:
  RtmpSenderImpl();

  int Initialize(const RtmpConfig& rtmp_config) override;
  int Start(const char* url) override;
  void Stop() override;
  void Release() override;
  int SendAudioFrame(const uint8_t* data, size_t size, int64_t pts_ms) override;
  int SendVideoFrame(const uint8_t* data, size_t size, int64_t pts_ms, bool is_key) override;
  int WriteFlvFile(const char* filename, bool is_have_audio = true,
                   bool is_have_video = true) override;
  void WriteFlvClose() override;
  int RegisterRtmpSenderObserver(IRtmpSenderObserver* observer) override;
  int UnregisterRtmpSenderObserver(IRtmpSenderObserver* observer) override;

 public:
  void OnRtmpStreamingBufferState(RtmpStreamingBufferState status) override;

 private:
  virtual ~RtmpSenderImpl();
  int OpenRtmp();
  int SendMetaData();
  void CloseRtmp();
  int SendPacketAsync();
  int TrySendPacket();
  int DoSendPacket(const RtmpMediaFrame& frame);
  void UpdateStateAndNotifyIfNeeded(RtmpSocketState state);
  void ReconnectRtmp();
  void StatisticTask();
  void CleanUpReconnection();
  bool NeedLogStream() { return send_frame_cnt_ <= 3 || commons::need_log(commons::LOG_STREAM); }

 private:
  srs_rtmp_t rtmp_handler_;
  RtmpConfig rtmp_config_;
  std::string publish_url_;
  utils::RtcSyncCallback<IRtmpSenderObserver>::Type sender_observers_;
  std::unique_ptr<RtmpStreamingBuffer> streaming_buffer_;
  bool is_started_;
  bool is_stopping_;
  std::atomic_bool is_sending_;
  bool is_reconnecting_;
  int reconnect_count_max_;
  int reconnect_count_current_;
  uint32_t relative_timestamps_;
  std::unique_ptr<commons::timer_base> statistic_timer_;
  std::unique_ptr<commons::timer_base> reconnect_timer_;
  RtmpSocketState current_socket_state_;
  utils::worker_type rtmp_sender_worker_;
  std::unique_ptr<async_queue_type> rtmp_sending_queue_;
  int send_frame_cnt_;
  int last_video_width_;
  int last_video_height_;

  struct STATISTIC_DATA {
    std::atomic_uint64_t last_send_video_bytes{0};
    std::atomic_uint64_t last_send_video_frame_count{0};
    std::atomic_uint64_t last_send_audio_bytes{0};
    std::atomic_uint64_t last_check_time{0};
    void Clear() {
      last_send_video_bytes.store(0);
      last_send_video_frame_count.store(0);
      last_send_audio_bytes.store(0);
      last_check_time.store(0);
    }
  };
  std::unique_ptr<STATISTIC_DATA> last_statistic_data_;
};

}  // namespace rtc
}  // namespace agora
